package com.gate.ws.spring.rocketmq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.gate.ws.spring.cache.LocalChannelCache;
import com.gate.ws.spring.cache.TempChannelCache;
import com.gate.ws.spring.constants.ChannelConstants;
import com.gate.ws.spring.utils.JsonUtils;
import com.nami.basic.bo.mq.DownLoginWrap;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import static com.nami.basic.constants.DownTopicConst.LOGIN_TOPIC;


@Slf4j
@Component
@RocketMQMessageListener(topic = LOGIN_TOPIC, consumerGroup = "${rocketmq.consumer.loginConsumerGroup}", selectorExpression = "${nami.mqTags}", consumeMode = ConsumeMode.CONCURRENTLY)
public class DownLoginConsumer implements RocketMQListener<String> {


    @Override
    public void onMessage(String msg) {
        try {
            TimeInterval timer = DateUtil.timer();
            DownLoginWrap downLoginWrap = JsonUtils.parseObject(msg, DownLoginWrap.class);

            if (downLoginWrap.getSuccess()) {
                log.info("[down-login] ws节点开始处理[{}]登录成功消息", downLoginWrap.getUserId());
                String tempNo = downLoginWrap.getTempFingerprint();
                Channel channel = TempChannelCache.getTempChannel(tempNo);
                if (channel == null) {
                    log.warn("[down-login] tempChannel is null。");
                    TempChannelCache.removeChannel(tempNo);
                    return;
                }
                bindingUserInfoToChannel(channel, downLoginWrap);
                LocalChannelCache.put(channel.id().asLongText(), downLoginWrap.getIdentityNo());
                LocalChannelCache.put(downLoginWrap.getIdentityNo(), channel);
                // 删除临时cache
                TempChannelCache.removeChannel(tempNo);
                channel.writeAndFlush(downLoginWrap.getPacket());
                log.info("[down-login] ws节点向客户端【{}】写入成功. use:{}ms", channel.remoteAddress(), timer.interval());
                return;
            }

            Channel tempChannel = TempChannelCache.getTempChannel(downLoginWrap.getTempFingerprint());
            if (tempChannel == null) {
                TempChannelCache.removeChannel(downLoginWrap.getTempFingerprint());
                return;
            }
            tempChannel.writeAndFlush(downLoginWrap.getPacket());
            TempChannelCache.removeChannel(downLoginWrap.getTempFingerprint());
            tempChannel.close();
        } catch (Exception var3) {
            var3.printStackTrace();
            log.error("[down-login] error:", var3.getMessage());
        }
    }

    private static void bindingUserInfoToChannel(Channel channel, DownLoginWrap wrap) {
        channel.attr(AttributeKey.valueOf(ChannelConstants.USER_TOKEN)).set(wrap.getToken());
        channel.attr(AttributeKey.valueOf(ChannelConstants.USER_ID)).set(wrap.getUserId());
        channel.attr(AttributeKey.valueOf(ChannelConstants.APP_KEY)).set(wrap.getAppKey());
        channel.attr(AttributeKey.valueOf(ChannelConstants.DEVICE_TYPE)).set(wrap.getDeviceType());
        channel.attr(AttributeKey.valueOf(ChannelConstants.IMEI)).set(wrap.getImei());
        channel.attr(AttributeKey.valueOf(ChannelConstants.READ_TIME)).set(System.currentTimeMillis());
    }


}
